/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package remoting;


import static common.constants.CommonConstants.ACCEPTS_KEY;
import static common.constants.CommonConstants.DEFAULT_ACCEPTS;
import static common.constants.CommonConstants.DEFAULT_IDLE_TIMEOUT;
import static common.constants.CommonConstants.IDLE_TIMEOUT_KEY;
import static common.constants.CommonConstants.THREADS_KEY;

import common.constants.CommonConstants;
import common.exception.RemotingException;
import common.model.Codec2;
import common.model.MagicUrl;
import common.util.ExecutorUtil;
import common.util.NetUtils;
import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import remoting.channel.Channel;
import remoting.channel.ChannelHandler;
import remoting.channel.ChannelHandlers;
import remoting.store.DataStore;
import remoting.store.DataStoreFactory;
import serialization.Serializer;
import threadpools.ThreadPool;

/**
 * AbstractClient
 */
public abstract class AbstractClient extends AbstractPeer implements Client {

  protected static final String CLIENT_THREAD_POOL_NAME = "MagicClientHandler";
  private static final Logger logger = LoggerFactory.getLogger(AbstractClient.class);
  private final Lock connectLock = new ReentrantLock();
  private final boolean needReconnect;
  protected volatile ExecutorService executor;
  private int accepts;
  private int idleTimeout;

  public AbstractClient(MagicUrl url, ChannelHandler handler, Codec2 codec)
      throws RemotingException {
    super(url, handler, codec);

    needReconnect = url.getParameter(CommonConstants.SEND_RECONNECT_KEY, false);
    this.accepts = url.getParameter(ACCEPTS_KEY, DEFAULT_ACCEPTS);
    this.idleTimeout = url.getParameter(IDLE_TIMEOUT_KEY, DEFAULT_IDLE_TIMEOUT);
    try {
      doOpen();
    } catch (Throwable t) {
      close();
      throw new RemotingException(url.toInetSocketAddress(), null,
          "Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
              + " connect to the server " + getRemoteAddress() + ", cause: " + t.getMessage(), t);
    }
    try {
      // connect.
      connect();
      if (logger.isInfoEnabled()) {
        logger.info("Start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
            + " connect to the server " + getRemoteAddress());
      }
    } catch (RemotingException t) {
      if (url.getParameter(CommonConstants.CHECK_KEY, true)) {
        close();
        throw t;
      } else {
        logger
            .warn("Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
                + " connect to the server " + getRemoteAddress()
                + " (check == false, ignore and retry later!), cause: " + t.getMessage(), t);
      }
    } catch (Throwable t) {
      close();
      throw new RemotingException(url.toInetSocketAddress(), null,
          "Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
              + " connect to the server " + getRemoteAddress() + ", cause: " + t.getMessage(), t);
    }
    DataStore dataStore = DataStoreFactory.getInstance();
    executor = (ExecutorService) dataStore
        .get(CommonConstants.CONSUMER_SIDE, Integer.toString(url.getPort()));
    dataStore.remove(CommonConstants.CONSUMER_SIDE, Integer.toString(url.getPort()));
  }

  protected static ChannelHandler wrapChannelHandler(MagicUrl url, ChannelHandler handler,
      ThreadPool threadPool) {
    url = ExecutorUtil.setThreadName(url, CLIENT_THREAD_POOL_NAME);
    url = url.addParameterIfAbsent(CommonConstants.THREADPOOL_KEY,
        CommonConstants.DEFAULT_CLIENT_THREADPOOL);
    return ChannelHandlers.wrap(handler, url, threadPool);
  }

  public InetSocketAddress getConnectAddress() {
    return new InetSocketAddress(NetUtils.filterLocalHost(getUrl().getHost()), getUrl().getPort());
  }

  @Override
  public InetSocketAddress getRemoteAddress() {
    Channel channel = getChannel();
    if (channel == null) {
      return getUrl().toInetSocketAddress();
    }
    return channel.getRemoteAddress();
  }

  @Override
  public InetSocketAddress getLocalAddress() {
    Channel channel = getChannel();
    if (channel == null) {
      return InetSocketAddress.createUnresolved(NetUtils.getLocalHost(), 0);
    }
    return channel.getLocalAddress();
  }

  @Override
  public boolean isConnected() {
    Channel channel = getChannel();
    if (channel == null) {
      return false;
    }
    return channel.isConnected();
  }

  @Override
  public Object getAttribute(String key) {
    Channel channel = getChannel();
    if (channel == null) {
      return null;
    }
    return channel.getAttribute(key);
  }

  @Override
  public void setAttribute(String key, Object value) {
    Channel channel = getChannel();
    if (channel == null) {
      return;
    }
    channel.setAttribute(key, value);
  }

  @Override
  public void removeAttribute(String key) {
    Channel channel = getChannel();
    if (channel == null) {
      return;
    }
    channel.removeAttribute(key);
  }

  @Override
  public boolean hasAttribute(String key) {
    Channel channel = getChannel();
    if (channel == null) {
      return false;
    }
    return channel.hasAttribute(key);
  }

  @Override
  public void sendMessage(Object message, boolean sent) throws RemotingException {
    if (needReconnect && !isConnected()) {
      connect();
    }
    Channel channel = getChannel();
    //TODO Can the value returned by getChannel() be null? need improvement.
    if (channel == null || !channel.isConnected()) {
      throw new RemotingException(this,
          "message can not send, because channel is closed . url:" + getUrl());
    }
    channel.sendMessage(message, sent);
  }

  protected void connect() throws RemotingException {

    connectLock.lock();

    try {

      if (isConnected()) {
        return;
      }

      doConnect();

      if (!isConnected()) {
        throw new RemotingException(this,
            "Failed connect to server " + getRemoteAddress() + " from " + getClass().getSimpleName()
                + " "
                + NetUtils.getLocalHost() + " using magic version " + CommonConstants.VERSION
                + ", cause: Connect wait timeout: " + getConnectTimeout() + "ms.");

      } else {
        if (logger.isInfoEnabled()) {
          logger.info("Succeed connect to server " + getRemoteAddress() + " from " + getClass()
              .getSimpleName() + " "
              + NetUtils.getLocalHost() + " using magic version " + CommonConstants.VERSION
              + ", channel is " + this.getChannel());
        }
      }

    } catch (RemotingException e) {
      throw e;

    } catch (Throwable e) {
      throw new RemotingException(this,
          "Failed connect to server " + getRemoteAddress() + " from " + getClass().getSimpleName()
              + " "
              + NetUtils.getLocalHost() + " using magic version " + CommonConstants.VERSION
              + ", cause: " + e.getMessage(), e);

    } finally {
      connectLock.unlock();
    }
  }

  public void disconnect() {
    connectLock.lock();
    try {
      try {
        Channel channel = getChannel();
        if (channel != null) {
          channel.close();
        }
      } catch (Throwable e) {
        logger.warn(e.getMessage(), e);
      }
      try {
        doDisConnect();
      } catch (Throwable e) {
        logger.warn(e.getMessage(), e);
      }
    } finally {
      connectLock.unlock();
    }
  }

  @Override
  public void reconnect() throws RemotingException {
    if (!isConnected()) {
      connectLock.lock();
      try {
        if (!isConnected()) {
          disconnect();
          connect();
        }
      } finally {
        connectLock.unlock();
      }
    }
  }

  @Override
  public void close() {

    try {
      super.close();
    } catch (Throwable e) {
      logger.warn(e.getMessage(), e);
    }

    try {
      if (executor != null) {
        ExecutorUtil.shutdownNow(executor, 100);
      }
    } catch (Throwable e) {
      logger.warn(e.getMessage(), e);
    }

    try {
      disconnect();
    } catch (Throwable e) {
      logger.warn(e.getMessage(), e);
    }

    try {
      doClose();
    } catch (Throwable e) {
      logger.warn(e.getMessage(), e);
    }
  }

  @Override
  public void close(int timeout) {
    ExecutorUtil.gracefulShutdown(executor, timeout);
    close();
  }

  @Override
  public String toString() {
    return getClass().getName() + " [" + getLocalAddress() + " -> " + getRemoteAddress() + "]";
  }

  /**
   * Open client.
   *
   * @throws Throwable
   */
  protected abstract void doOpen() throws Throwable;

  /**
   * Close client.
   *
   * @throws Throwable
   */
  protected abstract void doClose() throws Throwable;

  /**
   * Connect to server.
   *
   * @throws Throwable
   */
  protected abstract void doConnect() throws Throwable;

  /**
   * disConnect to server.
   *
   * @throws Throwable
   */
  protected abstract void doDisConnect() throws Throwable;

  /**
   * Get the connected channel.
   *
   * @return channel
   */
  protected abstract Channel getChannel();

  @Override
  public void reset(MagicUrl url) {
    if (url == null) {
      return;
    }
    try {
      if (url.hasParameter(ACCEPTS_KEY)) {
        int a = url.getParameter(ACCEPTS_KEY, 0);
        if (a > 0) {
          this.accepts = a;
        }
      }
    } catch (Throwable t) {
      logger.error(t.getMessage(), t);
    }
    try {
      if (url.hasParameter(IDLE_TIMEOUT_KEY)) {
        int t = url.getParameter(IDLE_TIMEOUT_KEY, 0);
        if (t > 0) {
          this.idleTimeout = t;
        }
      }
    } catch (Throwable t) {
      logger.error(t.getMessage(), t);
    }
    try {
      if (url.hasParameter(THREADS_KEY)
          && executor instanceof ThreadPoolExecutor && !executor.isShutdown()) {
        ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) executor;
        int threads = url.getParameter(THREADS_KEY, 0);
        int max = threadPoolExecutor.getMaximumPoolSize();
        int core = threadPoolExecutor.getCorePoolSize();
        if (threads > 0 && (threads != max || threads != core)) {
          if (threads < core) {
            threadPoolExecutor.setCorePoolSize(threads);
            if (core == max) {
              threadPoolExecutor.setMaximumPoolSize(threads);
            }
          } else {
            threadPoolExecutor.setMaximumPoolSize(threads);
            if (core == max) {
              threadPoolExecutor.setCorePoolSize(threads);
            }
          }
        }
      }
    } catch (Throwable t) {
      logger.error(t.getMessage(), t);
    }
    super.setUrl(getUrl().addParameters(url.getParameters()));
  }

  public int getAccepts() {
    return accepts;
  }

  public int getIdleTimeout() {
    return idleTimeout;
  }
}
